import 'dart:async';

import 'package:get/get.dart';
import 'package:meiyunapp/page/collection/mqtt_message_collection.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';
import 'package:typed_data/typed_buffers.dart';

final StreamController<String> mStreamController = StreamController.broadcast();

class MqttServices extends GetxService {
  late MqttServerClient client;

  Future<MqttServices> init() async {
    connect();
    return Future.value(this);
  }

  static MqttServices get to => Get.find();

  @override
  void onReady() {
    super.onReady();
    Get.put(MQTTMessageCollection(),
        tag: MQTTMessageCollection.TAG, permanent: true);
  }

  Future<MqttServerClient> connect() async {
    client = MqttServerClient('47.113.99.232', 'flutter01');
    client.port = 1883;
    client.logging(on: true);
    client.onConnected = onConnected;
    client.onDisconnected = onDisconnected;
    client.onSubscribed = onSubscribed;
    client.onSubscribeFail = onSubscribeFail;
    client.pongCallback = pong;
    client.setProtocolV311();
    final connMessage = MqttConnectMessage()
        .authenticateAs("user", "user888")
        .withWillTopic('willtopicu')
        .withWillMessage('Will message')
        .startClean()
        .withWillQos(MqttQos.atLeastOnce);
    client.connectionMessage = connMessage;
    client.setProtocolV311();
    try {
      await client.connect();
    } catch (e) {
      print('Exception: $e');
      client.disconnect();
    }

    client.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
      final message = c[0].payload as MqttPublishMessage;

      final payload =
          MqttPublishPayload.bytesToStringAsString(message.payload.message);

      print('Received message:$payload from topic: ${c[0].topic}>');

      Get.find<MQTTMessageCollection>(tag: MQTTMessageCollection.TAG)
          .sendMessage(payload, c[0].topic);

      mStreamController.add(payload);
    });
    return client;
  }

  //Subscribe
  void subscribe(String topic) {
    client.subscribe(topic, MqttQos.atLeastOnce);
  }

  void sendMessage(String topic, String data) {
    Uint8Buffer uint8buffer = Uint8Buffer();
    uint8buffer.addAll(data.codeUnits);
    client.publishMessage(topic, MqttQos.atLeastOnce, uint8buffer,
        retain: false);
  }

// connection succeeded
  void onConnected() {
    print('Connected连接成功');
    subscribe("admin");
  }

// unconnected
  void onDisconnected() {
    print('Disconnected连接失败');
  }

// subscribe to topic succeeded
  void onSubscribed(String topic) {
    print('Subscribed topic: $topic');
  }

// subscribe to topic failed
  void onSubscribeFail(String topic) {
    print('Failed to subscribe $topic');
  }

// unsubscribe succeeded
  void onUnsubscribed(String topic) {
    print('Unsubscribed topic: $topic');
  }

// PING response received
  void pong() {
    print('Ping response client callback invoked');
  }

  @override
  void onClose() {
    super.onClose();
    mStreamController.close();

    if (client != null) {
      client.disconnect();
    }
  }
}
